package stock

import (
	"time"

	"gitee.com/lyuanbo/stock-plus/internal/svc"
	"gitee.com/lyuanbo/stock-plus/pkg/consts"
	"gitee.com/lyuanbo/stock-plus/pkg/mathx"
	"gitee.com/lyuanbo/stock-plus/pkg/stock/barchecker"
	storagememory "gitee.com/lyuanbo/stock-plus/pkg/stock/storage_memory"
	"gitee.com/lyuanbo/stock-plus/pkg/stock/types"
	"gitee.com/lyuanbo/stock-plus/pkg/worker_manager"
	"github.com/alitto/pond/v2"
	"github.com/bytedance/sonic"
	"github.com/google/uuid"
	"github.com/jinzhu/copier"
	"github.com/nats-io/nats.go/jetstream"
	"github.com/redis/go-redis/v9"
	"github.com/shopspring/decimal"
	"go.uber.org/zap"
)

type WorkerStartReq struct {
	Symbol string `json:"symbol"`
	Period string `json:"period"`
	MsgId  string `json:"msgId"`
}

func init() {
	workermanager.Register(func(svcCtx *svc.ServiceContext, cache *storagememory.Cache) workermanager.Tasker {
		return NewSyncKlineWorker(svcCtx, cache)
	})
}

var _ workermanager.Tasker = (*SyncKlineWorker)(nil)

type SyncKlineWorker struct {
	svcCtx   *svc.ServiceContext
	customer jetstream.Consumer

	cache *storagememory.Cache
}

func NewSyncKlineWorker(svcCtx *svc.ServiceContext, cache *storagememory.Cache) *SyncKlineWorker {
	customer, _ := svcCtx.JetStream.CreateOrUpdateConsumer(svcCtx.Ctx, "default", jetstream.ConsumerConfig{
		Name:          "sync_kline",
		Durable:       "sync_kline",
		FilterSubject: consts.TopicRealTime,
		DeliverPolicy: jetstream.DeliverAllPolicy,
		AckPolicy:     jetstream.AckExplicitPolicy,
		MaxAckPending: 1,                // 关键：确保单条处理
		AckWait:       10 * time.Second, // 快速超时
		MaxDeliver:    3,                // 重试3次
	})
	s := &SyncKlineWorker{
		svcCtx:   svcCtx,
		customer: customer,
		cache:    cache,
	}

	return s
}

func (s *SyncKlineWorker) Name() string {
	return "sync_kline"
}

func (s *SyncKlineWorker) Init() {
}

func (s *SyncKlineWorker) Start() {
	go s.Run()
}

func (s *SyncKlineWorker) Run() {
	_, _ = s.customer.Consume(func(msg jetstream.Msg) {
		var kline types.Kline
		err := sonic.Unmarshal(msg.Data(), &kline)
		if err != nil {
			s.svcCtx.Logger.Warn(s.svcCtx.Ctx, "sync_kline worker unmarshal msg error", zap.Error(err))
			_ = msg.Nak()
			return
		}

		msgId := msg.Headers().Get("Nats-Msg-Id")
		if msgId == "" {
			uv7, _ := uuid.NewV7()
			msgId = uv7.String()
		}

		if kline.StockInfo.Symbol == "sz002624" {
			s.svcCtx.Logger.Info(s.svcCtx.Ctx, "realtime", zap.String("msgId", msgId), zap.String("name", kline.StockInfo.Name), zap.String("period", "realtime"), zap.Any("kline", kline.KlineInfo))
		}

		defer func() {
			if kline.StockInfo.Symbol == "sz002624" {
				s.svcCtx.Logger.Info(s.svcCtx.Ctx, "----------------", zap.String("msgId", msgId))
			}
		}()

		pool := pond.NewPool(len(s.svcCtx.Conf.Get().Kline))
		for _, period := range s.svcCtx.Conf.Get().Kline {
			pool.Submit(func() {
				pipe := s.svcCtx.Redis.Pipeline()
				key := s.cache.GetKey(kline.StockInfo.Symbol, period)
				rb := s.cache.GetRingBuffer(key)

				rb.Mux.Lock()
				defer rb.Mux.Unlock()

				// 获取上一条数据
				lastKline, _ := rb.GetLatest()
				var lastMember string
				if lastKline != nil {
					lastM, _ := lastKline.KlineInfo.MarshalArray()
					lastMember = string(lastM)
				}

				if period == "60m" && kline.StockInfo.Symbol == "sz002624" {
					s.svcCtx.Logger.Info(s.svcCtx.Ctx, "lastKline", zap.String("msgId", msgId), zap.String("name", lastKline.StockInfo.Name), zap.String("period", period), zap.Any("kline", lastKline.KlineInfo))
				}

				// 计算时间
				barTime, isNewBar := barchecker.Check(lastKline.KlineInfo.Timestamp, kline.KlineInfo.Timestamp, kline.StockInfo.Exchange.Code, period)
				if period == "60m" && kline.StockInfo.Symbol == "sz002624" && lastKline != nil {
					s.svcCtx.Logger.Info(s.svcCtx.Ctx, "barTime", zap.String("msgId", msgId), zap.String("name", lastKline.StockInfo.Name), zap.String("period", period), zap.Time("barTime", barTime),
						zap.Any("kline", kline.KlineInfo.Timestamp), zap.Any("lastKline", lastKline.KlineInfo.Timestamp), zap.Bool("isNewBar", isNewBar))
				}
				if isNewBar {
					var tKline types.Kline
					_ = copier.Copy(&tKline, &kline)

					tKline.KlineInfo.Timestamp = barTime
					rb.PushBack(&tKline)
					buf, _ := tKline.KlineInfo.MarshalArray()
					pipe.ZAdd(s.svcCtx.Ctx, key, redis.Z{Score: float64(barTime.Unix()), Member: buf})
					if period == "60m" && tKline.StockInfo.Symbol == "sz002624" {
						s.svcCtx.Logger.Info(s.svcCtx.Ctx, "push back", zap.String("msgId", msgId), zap.String("name", lastKline.StockInfo.Name), zap.String("period", period), zap.Any("kline", tKline.KlineInfo))
					}
					return
				}

				// 不需要添加bar
				lastKline.KlineInfo.Close = kline.KlineInfo.Close
				lastKline.KlineInfo.High = mathx.MaxDecimal(lastKline.KlineInfo.High, kline.KlineInfo.High)
				lastKline.KlineInfo.Low = mathx.MinDecimal(lastKline.KlineInfo.Low, kline.KlineInfo.Low)
				lastKline.KlineInfo.Volume += kline.KlineInfo.Volume
				lastKline.KlineInfo.Amount = lastKline.KlineInfo.Amount.Add(lastKline.KlineInfo.Amount)

				// 获取倒数第2个数据
				tmpKline := rb.GetIndex(-2)
				if err == nil && tmpKline != nil {
					lastKline.KlineInfo.TurnoverRate = kline.KlineInfo.TurnoverRate // 换手率没有办法根据实时数据计算
					lastKline.KlineInfo.Amplitude = tmpKline.KlineInfo.Close.Sub(kline.KlineInfo.Close).Div(tmpKline.KlineInfo.Close).Mul(decimal.NewFromInt(100))
					lastKline.KlineInfo.Change = lastKline.KlineInfo.High.Sub(lastKline.KlineInfo.Low)
					lastKline.KlineInfo.ChangePercent = lastKline.KlineInfo.Change.Div(lastKline.KlineInfo.High).Mul(decimal.NewFromInt(100))
				}

				pipe.ZRem(s.svcCtx.Ctx, key, lastMember)
				buf, _ := lastKline.KlineInfo.MarshalArray()
				pipe.ZAdd(s.svcCtx.Ctx, key, redis.Z{Score: float64(barTime.Unix()), Member: buf})

				pipe.Expire(s.svcCtx.Ctx, key, 24*time.Hour)
				if period == "60m" && kline.StockInfo.Symbol == "sz002624" {
					s.svcCtx.Logger.Info(s.svcCtx.Ctx, "update", zap.String("msgId", msgId), zap.String("name", lastKline.StockInfo.Name), zap.String("period", period), zap.Any("kline", lastKline.KlineInfo))
				}
				_, err = pipe.Exec(s.svcCtx.Ctx)
				if err != nil {
					s.svcCtx.Logger.Warn(s.svcCtx.Ctx, "cci worker exec redis pipeline error", zap.Error(err))
				}

				w := WorkerStartReq{
					Symbol: kline.StockInfo.Symbol,
					Period: period,
					MsgId:  msgId,
				}
				buf, _ = sonic.Marshal(w)
				_ = s.svcCtx.Nats.Publish(consts.TopicWorkerStart, buf)
			})
		}
		pool.StopAndWait()
		_ = msg.Ack()
	})
}
